package com.wudl.utils

import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.kafka.KafkaUtils


/**
 * 读取kafka 的工具类
 */
object MyKafkaUtil {

  //从kafka读数据
  private val params = Map(
    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG ->"node01.com:6667",
    ConsumerConfig.GROUP_ID_CONFIG->  "wudl001"
  )

  def  getKafkaStream(ssc:StreamingContext,topic:String,otherTopic:String*): DStream[String] ={
    KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,
      params,
      (otherTopic :+ topic).toSet).map(_._2)
  }

}
